package com.inspur.cloud.service.dataspace;

import com.inspur.cloud.configuration.AmbariConfig;
import com.inspur.cloud.enums.PermissionEnum;
import com.inspur.cloud.enums.ResourceTypeEnum;
import com.inspur.cloud.util.TConstants;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Service;

import java.util.Locale;

@Service
public class InitializeService {
    private static final Logger logger = LoggerFactory.getLogger(InitializeService.class);

    @Autowired
    KerberosService kerberosService;
    @Autowired
    RangerService rangerService;
    @Autowired
    RangerRestService rangerRestService;
    @Autowired
    LinuxService linuxService;
    @Autowired
    HdfsService hdfsService;
    @Autowired
    AmbariConfig ambariConfig;
    @Autowired
    ClusterConfigUtil clusterConfigUtil;

    /**
     * 判断dataspace服务是否进行了初始化
     * 如果开启安全认证，判断是否有dataspace票据信息，如果没有，则说明没有进行初始化
     * 上述逻辑是为了解决集群由非安全状态切换到安全状态时，系统没有dataspace票据的问题
     * 如果某个服务的ranger plugin未开启，则不进行判断
     * 如果开启后，最大权限策略中无dataspace用户，则说明未进行初始化
     * @return
     */
    public boolean checkInitialized() {
        logger.info("dataspace service begin to check initial status.");
        //如果集群开启安全认证，判断是否有dataspace票据信息
        if (clusterConfigUtil.checkKerberized()) {
            String dataspacePrincipal = TConstants.DATASPACE_PRINCIPAL;
            if (!kerberosService.findPrincipalExists(dataspacePrincipal)) {
                return false;
            }
        }
        String user = TConstants.DATASPACE_USER;
        for (ResourceTypeEnum e : ResourceTypeEnum.values()) {
            String type = e.getResourceType().toLowerCase(Locale.CHINA);
            //判断是否开启对应的ranger的Plugin
            if (!checkRangerPluginOpen(type)) {
                continue;
            }
            //判断最大权限中是否有dataspace用户
            switch (type) {
                case "hdfs":
                    String hdfsMaxResource = "/*";
                    if (!rangerRestService.isResourceContainUser("hdfs",hdfsMaxResource,user)) {
                        logger.info("dataspace service isInitialized check ["+ type +"] have not been authorized.");
                        return false;
                    }
                    break;
                case "hbase":
                    String hbaseMaxResource = "*|*|*";
                    if (!rangerRestService.isResourceContainUser("hbase",hbaseMaxResource,user)) {
                        logger.info("dataspace service isInitialized check ["+ type +"] have not been authorized.");
                        return false;
                    }
                    break;
                case "hive":
                    String hiveMaxResource = "*|*|*";
                    if (!rangerRestService.isResourceContainUser("hive",hiveMaxResource,user)) {
                        logger.info("dataspace service isInitialized check ["+ type +"] have not been authorized.");
                        return false;
                    }
                    break;
                case "yarn":
                    String yarnMaxResource = "*";
                    if (!rangerRestService.isResourceContainUser("yarn",yarnMaxResource,user)) {
                        logger.info("dataspace service isInitialized check ["+ type +"] have not been authorized.");
                        return false;
                    }
                    break;
                case "kafka":
                    String kafkaMaxResource = "*";
                    if (!rangerRestService.isResourceContainUser("kafka",kafkaMaxResource,user)) {
                        logger.info("dataspace service isInitialized check ["+ type +"] have not been authorized.");
                        return false;
                    }
                    break;
            }
        }
        return true;
    }

    /**
     * 初始化操作
     * 所有节点创建dataspace用户并赋最大权限
     */
    public void initial() {
        //如果开了安全，但没有dataspace票据，则生成
        if (clusterConfigUtil.checkKerberized()) {
            String dataspacePrincipal = TConstants.DATASPACE_PRINCIPAL;
            if (!kerberosService.findPrincipalExists(dataspacePrincipal)) {
                logger.info("dataspace service begin to initial dataspace kerberos user ["+ dataspacePrincipal +"].");
                kerberosService.addDataspacePrincipal(dataspacePrincipal);
            }
        }
        logger.info("dataspace service begin to initial and set ranger privileges.");
        //在各个主机节点上创建dataspace用户、组，密码也一样
        String user = TConstants.DATASPACE_USER;
        linuxService.operateUerAndGroupOnLinux(user,"hdfs",user,1);
        //rangerUserSync节点同步用户
        rangerService.rangerUserSync(user,user);
        //ranger中增加权限
        for (ResourceTypeEnum e : ResourceTypeEnum.values()) {
            String type = e.getResourceType().toLowerCase(Locale.CHINA);
            //判断是否开启对应的ranger的Plugin
            if (!checkRangerPluginOpen(type)) {
                continue;
            }
            switch (type) {
                case "hdfs":
                    logger.info("dataspace service initial setPermissionInRanger for ["+ type +"] service.");
                    rangerService.setPermissionInRanger(type,"/*", PermissionEnum.getAllPermissionByType(ResourceTypeEnum.HDFS.getType()),user);
                    break;
                case "hbase":
                    logger.info("dataspace service initial setPermissionInRanger for ["+ type +"] service.");
                    rangerService.setPermissionInRanger(type,"*|*|*", PermissionEnum.getAllPermissionByType(ResourceTypeEnum.HBASE.getType()),user);
                    break;
                case "hive":
                    logger.info("dataspace service initial setPermissionInRanger for ["+ type +"] service.");
                    rangerService.setPermissionInRanger(type,"*|*|*", PermissionEnum.getAllPermissionByType(ResourceTypeEnum.HIVE.getType()),user);
                   //初始化时，在yarn的all-queue策略中增加hive用户
                    if (checkRangerPluginOpen("yarn")) {
                        logger.info("dataspace service initial setPermissionInRanger for yarn all-queue add hive user.");
                        rangerService.setPermissionInRanger("yarn","*",PermissionEnum.getAllPermissionByType(ResourceTypeEnum.YARN.getType()),"hive");
                    }
                    //初始化时，在hdfs的all-path策略中增加hive用户
                    if (checkRangerPluginOpen("hdfs")) {
                        logger.info("dataspace service initial setPermissionInRanger for hdfs all-path add hive user.");
                        rangerService.setPermissionInRanger("hdfs","/*",PermissionEnum.getAllPermissionByType(ResourceTypeEnum.HDFS.getType()),"hive");
                    }
                    break;
                case "yarn":
                    logger.info("dataspace service initial setPermissionInRanger for ["+ type +"] service.");
                    rangerService.setPermissionInRanger(type,"*", PermissionEnum.getAllPermissionByType(ResourceTypeEnum.YARN.getType()),user);
                    break;
                case "kafka":
                    //初始化时,如果开kerberos了，则去掉alltopic策略的public组
                    if (clusterConfigUtil.checkKerberized()) {
                        logger.info("dataspace service initial formatKafkaAllTopicPolicy for remove public group.");
                        rangerService.formatKafkaAllTopicPolicy();
                    }
                    logger.info("dataspace service initial setPermissionInRanger for ["+ type +"] service.");
                    String permissions = PermissionEnum.getAllPermissionByType(ResourceTypeEnum.KAFKA.getType());
                    if (!permissions.contains("kafka_admin")) {
                        permissions += "|kafka_admin";
                    }
                    rangerService.setPermissionInRanger(type,"*", permissions,user);
                    break;
            }
        }
        //初始化时对hdfs的/user目录权限进行设置
        try {
            if (ambariConfig.isSetHdfsUserDirMode()) {
                logger.info("dataspace service initialize and set HDFS user dir to 777 mode.");
                hdfsService.setPermission("/user");
            }
        }catch (Exception e) {
            logger.info("dataspace service set HDFS user dir to 777 mode error and other operations will continue.");
        }
    }


    /**
     * 初始化时检测判断某个服务的plugin是否打开
     * @param type
     * @return
     */
    public boolean checkRangerPluginOpen(String type) {
        //判断是否开启对应的ranger的Plugin
        String rangerServiceName = rangerRestService.getServiceName(type);
        if ("".equals(rangerServiceName)) {
            logger.info("dataspace service checkRangerPluginOpen ["+ type +"] have not plugin.");
            return false;
        }
        return true;
    }
}
